package com.pactera.rabbitmq.autoconfigure;

import com.google.common.collect.Lists;
import com.pactera.rabbitmq.service.consumer.RtdsAccept;
import com.pactera.rabbitmq.service.consumer.RtdsDefaultSubscriber;
import com.pactera.rabbitmq.service.consumer.RtdsSubscriber;
import com.pactera.rabbitmq.utils.AdminUtils;
import com.pactera.rabbitmq.utils.RabbitUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;

/**
 * @author: liqingdian
 **/
@Configuration
@AutoConfigureAfter(RabbitAutoConfiguration.class)
@EnableConfigurationProperties({RabbitmqProperties.class})
@Import({RabbitmqConfiguration.class})
public class RabbitmqAutoConfiguration {
    private static final Logger log = LoggerFactory.getLogger(RabbitmqAutoConfiguration.class);

    @Autowired
    private RabbitmqProperties rabbitmqProperties;

    @Bean
    @ConditionalOnBean(name = "amqpAdmin")
    @ConditionalOnMissingBean
    public RabbitUtils rabbitUtils(AmqpAdmin amqpAdmin) {
        RabbitUtils builder = RabbitUtils.builder(amqpAdmin);
        log.info("rabbitmq-RabbitUtils初始化成功");
        return builder;
    }

    @Bean
    @ConditionalOnMissingBean
    @ConditionalOnBean(name = "restTemplate")
    @ConditionalOnProperty(
            prefix = "spring.rabbitmq.pactera.management",
            name = {"enabled"},
            havingValue = "true")
    public AdminUtils adminUtils(RestTemplate restTemplate) {
        AdminUtils builder = AdminUtils.builder(restTemplate, rabbitmqProperties.getManagement());
        log.info("rabbitmq-AdminUtils初始化成功");
        return builder;
    }

    @Bean
    @ConditionalOnBean(name = {"rabbitTemplate"})
    @ConditionalOnSingleCandidate(RtdsAccept.class)
    @ConditionalOnMissingBean
    public RtdsSubscriber rtdsSubscriber(RtdsAccept rtdsAccept, RabbitTemplate rabbitTemplate) {
        return new RtdsDefaultSubscriber(rtdsAccept, rabbitTemplate);
    }

    @Bean
    @ConditionalOnBean(name = {"rtdsSubscriber", "rabbitTemplate"})
    @ConditionalOnMissingBean
    public SimpleMessageListenerContainer rtdsMessageListenerContainer(RtdsSubscriber rtdsSubscriber, RabbitTemplate rabbitTemplate) throws AmqpException, IOException {
        ConnectionFactory connectionFactory = rabbitTemplate.getConnectionFactory();
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        String[] queues = rabbitmqProperties.getQueues();
        if (ArrayUtils.isEmpty(queues)) {
            queues = StringUtils.toStringArray(Lists.newArrayList(connectionFactory.getUsername()));
        }
        container.setQueueNames(queues);//同时监听多个队列
        container.setExposeListenerChannel(true);//设置监听外露
        container.setDefaultRequeueRejected(false);//设置是否重回队列
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);//设置确认模式为自动确认
        container.setPrefetchCount(30);// 预取消息数
        container.setMessageListener(rtdsSubscriber);//监听处理类
        log.info("Rabbit队列{}- 实时数据同步监听-初始化成功", ArrayUtils.toString(queues));
        return container;
    }
}
